package com.jhf.youke.pulsar.domain.service;

import lombok.extern.log4j.Log4j2;
import org.apache.pulsar.client.admin.PulsarAdmin;
import org.apache.pulsar.common.policies.data.TenantInfo;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.HashSet;

/**
* @Description:
* @Param:
* @return:
* @Author: RHJ
* @Date: 2022/11/18
*/
@Component
@Log4j2
public class TenantService {

    @Resource
    private PulsarConfig pulsarConfig;

    private PulsarAdmin init() throws Exception{
        PulsarAdmin pulsarAdmin = PulsarAdmin.builder().serviceHttpUrl(pulsarConfig.getServiceHttpUrl()).build();

        return pulsarAdmin;
    }

    public void createTenant(String tenantName, String clusterName) throws Exception{
        PulsarAdmin pulsarAdmin = init();
        HashSet<String> clusters = new HashSet<>();
        clusters.add(clusterName);
        TenantInfo config = TenantInfo.builder().allowedClusters(clusters).build();
        pulsarAdmin.tenants().createTenant(tenantName,config);
        log.info("tenants {}", pulsarAdmin.tenants().getTenants());
        pulsarAdmin.close();
    }

    public void getTenants()throws Exception{
        PulsarAdmin pulsarAdmin = init();
        log.info("tenants {}", pulsarAdmin.tenants().getTenants());
    }

    public void createNameSpace(String nameSpace, String tenantName) throws Exception{
        PulsarAdmin pulsarAdmin = init();
        String ns = tenantName + "/" + nameSpace;
        pulsarAdmin.namespaces().createNamespace(ns);
        log.info("namespaces {}", pulsarAdmin.namespaces().getNamespaces(tenantName));
        pulsarAdmin.close();
    }

    public void getNameSpace(String tenantName)throws Exception{
        PulsarAdmin pulsarAdmin = init();
        log.info("namespaces {}", pulsarAdmin.namespaces().getNamespaces(tenantName));
    }

    public void createTopic(String nameSpace, String tenantName,String topic) throws Exception{
        PulsarAdmin pulsarAdmin = init();
        String name ="persistent://" + tenantName + "/" + nameSpace +"/" + topic;
        log.info("topic {}", name);
        pulsarAdmin.topics().createNonPartitionedTopic(name);
        log.info("topics {}", pulsarAdmin.topics().getList(tenantName + "/"+ nameSpace));
        pulsarAdmin.close();
    }

    public static void main(String[] args) throws Exception{
        TenantService tenantService = new TenantService();
        tenantService.createTenant("hczh","standalone");

    }


}
